package com.clear.subject.infra.basic.es;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class EsRestClient {

    // <集群名称, ES高级客户端>
    public static Map<String, RestHighLevelClient> clientMap = new HashMap<>();

    @Resource
    private EsConfigProperties esConfigProperties;

    private static final RequestOptions COMMON_OPTIONS;

    static {
        COMMON_OPTIONS = RequestOptions.DEFAULT.toBuilder().build();
    }


    @PostConstruct  // 在构造器之后执行
    public void initialize() {
        List<EsClusterConfig> esConfigs = esConfigProperties.getEsConfigs();
        // ===============遍历es配置，创建ES客户端=======================
        for (EsClusterConfig esConfig : esConfigs) {
            log.info("EsRestClient.initialize.config.name:{},node:{}", esConfig.getName(), esConfig.getNodes());

            // 初始化ES高级客户端，只要初始化成功，就put进clientMap
            RestHighLevelClient restHighLevelClient = initRestClient(esConfig);
            if (restHighLevelClient != null) {
                clientMap.put(esConfig.getName(), restHighLevelClient);
            } else {
                log.error("EsRestClient.initialize.config.name:{},node:{}.initError", esConfig.getName(), esConfig.getNodes());
            }
        }
    }

    /**
     * 初始化ES客户端
     *
     * @param esClusterConfig es集群配置
     * @return ES高级客户端
     */
    private RestHighLevelClient initRestClient(EsClusterConfig esClusterConfig) {
        // todo 我们定义以 , 分隔符分割集群内多个node
        String[] ipPortAddr = esClusterConfig.getNodes().split(",");

        List<HttpHost> httpHostList = new ArrayList<>(ipPortAddr.length);

        for (String ipPort : ipPortAddr) {
            String[] ipPortInfo = ipPort.split(":");
            if (ipPortInfo.length == 2) {
                // 配置正确，接着处理
                String host = ipPortInfo[0];
                Integer port = NumberUtils.toInt(ipPortInfo[1]);

                HttpHost httpHost = new HttpHost(host, port);
                httpHostList.add(httpHost);
            }
        }

        HttpHost[] httpHosts = new HttpHost[httpHostList.size()];
        httpHostList.toArray(httpHosts);

        RestClientBuilder builder = RestClient.builder(httpHosts);
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(builder);

        return restHighLevelClient;
    }

    /**
     * 根据集群名称获取RestHighLevelClient
     *
     * @param clusterName 集群名称
     * @return ES高级客户端
     */
    private static RestHighLevelClient getClient(String clusterName) {
        return clientMap.get(clusterName);
    }


    /**
     * 新增文档
     * POST http://192.168.188.150:9200/indexName/_doc/docId
     *
     * @param esIndexInfo  ES索引类 （集群名称，索引名称）
     * @param esSourceData （文档id，数据）
     * @return
     */
    public static boolean insertDoc(EsIndexInfo esIndexInfo, EsSourceData esSourceData) {
        try {
            // 1.准备Request
            IndexRequest indexRequest = new IndexRequest(esIndexInfo.getIndexName()).id(esSourceData.getDocId());
            // 2.准备DSL
            indexRequest.source(esSourceData.getData());
            // 3.发出请求
            RestHighLevelClient restHighLevelClient = getClient(esIndexInfo.getClusterName());
            IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            // 4.解析结果
            int status = indexResponse.status().getStatus();
            return status > 0;
        } catch (IOException e) {
            log.error("EsRestClient.insertDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 更新文档
     * UPDATE http://192.168.188.150:9200/indexName/_doc/docId
     *
     * @param esIndexInfo  ES索引类 （集群名称，索引名称）
     * @param esSourceData （文档id，数据）
     * @return
     */
    public static boolean updateDoc(EsIndexInfo esIndexInfo, EsSourceData esSourceData) {
        try {
            // 1.准备Request
            UpdateRequest updateRequest = new UpdateRequest();
            // 2.准备DSL
            updateRequest.index(esIndexInfo.getIndexName())
                    .id(esSourceData.getDocId())
                    .doc(esSourceData.getData());
            // 3.发出请求
            RestHighLevelClient restHighLevelClient = getClient(esIndexInfo.getClusterName());
            UpdateResponse updateResponse = restHighLevelClient.update(updateRequest, COMMON_OPTIONS);
            // 4.解析结果
            int status = updateResponse.status().getStatus();
            log.error("EsRestClient.updateDoc.status.size:{}", status);
            return status > 0;
        } catch (IOException e) {
            log.error("EsRestClient.updateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 根据查询更新文档
     *
     * @param esIndexInfo  ES索引类 （集群名称，索引名称）
     * @param queryBuilder
     * @param script       ES脚本
     * @param batchSize
     * @return
     */
    public static boolean updateByQuery(EsIndexInfo esIndexInfo, QueryBuilder queryBuilder, Script script, int batchSize) {
        if (log.isInfoEnabled()) {
            log.info("EsRestClient.updateByQuery.indexName:{}", esIndexInfo.getIndexName());
        }
        try {
            // 1.准备Request
            UpdateByQueryRequest updateByQueryRequest = new UpdateByQueryRequest(esIndexInfo.getIndexName());
            // 2.准备DSL
            updateByQueryRequest
                    .setQuery(queryBuilder)
                    .setScript(script)
                    .setBatchSize(batchSize);
            updateByQueryRequest.setAbortOnVersionConflict(false);
            // 3.发出请求
            RestHighLevelClient restHighLevelClient = getClient(esIndexInfo.getClusterName());
            BulkByScrollResponse response = restHighLevelClient.updateByQuery(updateByQueryRequest, RequestOptions.DEFAULT);
            // 4.解析结果
            List<BulkItemResponse.Failure> failures = response.getBulkFailures();
        } catch (Exception e) {
            log.error("EsRestClient.updateByQuery.exception:{}", e.getMessage(), e);
        }
        return true;
    }


    /**
     * 更新文档（批量）
     * UPDATE http://192.168.188.150:9200/indexName/_doc/docId
     *
     * @param esIndexInfo      ES索引类 （集群名称，索引名称）
     * @param esSourceDataList （文档id，数据）
     * @return
     */
    public static boolean batchUpdateDoc(EsIndexInfo esIndexInfo, List<EsSourceData> esSourceDataList) {
        try {
            boolean flag = false;
            // 1.准备Request
            BulkRequest bulkRequest = new BulkRequest();
            // 2.准备DSL
            for (EsSourceData esSourceData : esSourceDataList) {
                String docId = esSourceData.getDocId();
                if (StringUtils.isBlank(docId)) {
                    UpdateRequest updateRequest = new UpdateRequest()
                            .index(esIndexInfo.getIndexName())
                            .id(docId)
                            .doc(esSourceData.getData());
                    bulkRequest.add(updateRequest);
                    flag = true;
                }
            }
            // 3.发出请求
            // 4.解析结果
            RestHighLevelClient restHighLevelClient = getClient(esIndexInfo.getClusterName());
            if (flag) {
                BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, COMMON_OPTIONS);
                if (bulkResponse.hasFailures()) {
                    return false;
                }
            }
        } catch (IOException e) {
            log.error("EsRestClient.batchUpdateDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }


    /**
     * 删除文档
     *
     * @param esIndexInfo  ES索引类 （集群名称，索引名称）
     * @param esSourceData （文档id，数据）
     * @return
     */
    public static boolean delete(EsIndexInfo esIndexInfo, EsSourceData esSourceData) {
        try {
            // 1.准备Request
            DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(esIndexInfo.getIndexName());
            // 2.准备DSL
            deleteByQueryRequest.setQuery(QueryBuilders.matchAllQuery());  // 匹配所有文档
            // 3.发出请求
            RestHighLevelClient restHighLevelClient = getClient(esIndexInfo.getClusterName());
            BulkByScrollResponse bulkByScrollResponse = restHighLevelClient.deleteByQuery(deleteByQueryRequest, COMMON_OPTIONS);
            // 4.解析结果
            long deleted = bulkByScrollResponse.getDeleted();
            log.error("EsRestClient.delete.deleted.size:{}", deleted);
            return deleted > 0;
        } catch (IOException e) {
            log.error("EsRestClient.delete.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 匹配新增文档
     *
     * @param esIndexInfo      ES索引类 （集群名称，索引名称）
     * @param esSourceDataList
     * @return
     */
    public static boolean batchInsertDoc(EsIndexInfo esIndexInfo, List<EsSourceData> esSourceDataList) {
        if (log.isInfoEnabled()) {
            log.info("EsRestClient.batchInsertDoc.indexName:{}", esIndexInfo.getIndexName());
            log.info("EsRestClient.batchInsertDoc.批量新增ES:{}", esSourceDataList.size());
        }
        try {
            boolean flag = false;
            // 1.准备Request
            BulkRequest bulkRequest = new BulkRequest();
            // 2.准备DSL
            for (EsSourceData source : esSourceDataList) {
                String docId = source.getDocId();
                if (StringUtils.isNotBlank(docId)) {
                    IndexRequest indexRequest = new IndexRequest(esIndexInfo.getIndexName());
                    indexRequest.id(docId).source(source.getData());

                    bulkRequest.add(indexRequest);
                    flag = true;
                }
            }
            // 3.发出请求
            if (flag) {
                BulkResponse response = getClient(esIndexInfo.getClusterName()).bulk(bulkRequest, COMMON_OPTIONS);
                if (response.hasFailures()) {
                    return false;
                }
            }
        } catch (Exception e) {
            log.error("EsRestClient.batchInsertDoc.exception:{}", e.getMessage(), e);
        }
        return true;
    }


    /**
     * 删除文档
     *
     * @param esIndexInfo ES索引类 （集群名称，索引名称）
     * @param docId       文档id
     * @return
     */
    public static boolean deleteDoc(EsIndexInfo esIndexInfo, String docId) {
        try {
            // 1.准备Request
            DeleteRequest deleteRequest = new DeleteRequest();
            // 2.准备DSL
            deleteRequest.index(esIndexInfo.getIndexName()).id(docId);
            deleteRequest.timeout("2s");
            // 3.发出请求
            RestHighLevelClient restHighLevelClient = getClient(esIndexInfo.getClusterName());
            DeleteResponse delete = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            // 4.解析结果
            int status = delete.status().getStatus();
            return status > 0;
        } catch (IOException e) {
            log.error("EsRestClient.deleteDoc.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 根据文档id判断文档是否存在
     *
     * @param esIndexInfo ES索引类 （集群名称，索引名称）
     * @param docId       文档id
     * @return
     */
    public static boolean isExistDocById(EsIndexInfo esIndexInfo, String docId) {
        try {
            // 1.准备Request
            GetRequest getRequest = new GetRequest();
            // 2.准备DSL
            getRequest.index(esIndexInfo.getIndexName()).id(docId);
            // 3.发出请求
            RestHighLevelClient restHighLevelClient = getClient(esIndexInfo.getClusterName());
            boolean exists = restHighLevelClient.exists(getRequest, COMMON_OPTIONS);
            // 4.解析结果
            return exists;
        } catch (IOException e) {
            log.error("EsRestClient.isExistDocById.exception:{}", e.getMessage(), e);
        }
        return false;
    }

    /**
     * 根据文档id查询文档
     *
     * @param esIndexInfo ES索引类 （集群名称，索引名称）
     * @param docId       文档id
     * @return
     */
    public static Map<String, Object> getDocById(EsIndexInfo esIndexInfo, String docId) {
        try {
            // 1.准备Request
            GetRequest getRequest = new GetRequest();
            // 2.准备DSL
            getRequest.index(esIndexInfo.getIndexName()).id(docId);
            // 3.发出请求
            RestHighLevelClient restHighLevelClient = getClient(esIndexInfo.getClusterName());
            GetResponse getResponse = restHighLevelClient.get(getRequest, COMMON_OPTIONS);
            // 4.解析请求
            Map<String, Object> source = getResponse.getSource();
            return source;
        } catch (IOException e) {
            log.error("EsRestClient.getDocById.exception:{}", e.getMessage(), e);
        }
        return null;
    }

    /**
     * 精确查询
     *
     * @param esIndexInfo     ES索引类 （集群名称，索引名称）
     * @param esSearchRequest ES请求类
     * @return
     */
    public static SearchResponse searchWithTermQuery(EsIndexInfo esIndexInfo,
                                                     EsSearchRequest esSearchRequest) {
        try {
            BoolQueryBuilder boolQuery = esSearchRequest.getBq();  // bool查询
            String[] fields = esSearchRequest.getFields();  // 查询字段
            int from = esSearchRequest.getFrom();
            int size = esSearchRequest.getSize();
            Boolean needScroll = esSearchRequest.getNeedScroll();  // 需要快照
            Long minutes = esSearchRequest.getMinutes();    // 快照缓存时间
            String sortName = esSearchRequest.getSortName();    // 排序字段
            SortOrder sortOrder = esSearchRequest.getSortOrder();  // 排序类型
            HighlightBuilder highlightBuilder = esSearchRequest.getHighlightBuilder();  // 高亮

            // 1.准备Request
            SearchRequest searchRequest = new SearchRequest(esIndexInfo.getIndexName());
            // 2.准备DSL
            // 构建搜索条件
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

            searchSourceBuilder.query(boolQuery);
            searchSourceBuilder.fetchSource(fields, null).from(from).size(size);
            if (StringUtils.isNotBlank(sortName) && StringUtils.isNotBlank(sortOrder.toString())) {
                searchSourceBuilder.sort(sortName, sortOrder);
            }
            if (Objects.nonNull(highlightBuilder)) {
                searchSourceBuilder.highlighter(highlightBuilder);
            }
            searchSourceBuilder.timeout(new TimeValue(5, TimeUnit.SECONDS));    // 超时时间

            searchRequest.source(searchSourceBuilder).searchType(SearchType.DEFAULT);
            if (esSearchRequest.getNeedScroll()) {
                searchRequest.scroll(new TimeValue(minutes));   // 设置快照
            }
            // 3.发出请求
            RestHighLevelClient restHighLevelClient = getClient(esIndexInfo.getClusterName());
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, COMMON_OPTIONS);
            // 4.解析请求
            return searchResponse;
        } catch (IOException e) {
            log.error("EsRestClient.searchWithTermQuery.exception:{}", e.getMessage(), e);
        }
        return null;
    }


}
